/**
 * @(#)WriterThread.java 1.0 2014/02/14
 *
 * Copyright (c) 2014 SinoSoft Refactored by filon51.
 * All rights reserved.
 *
 */
package com.sinosoft.cloud.dp.thread;

//import EDU.oswego.cs.dl.util.concurrent.BoundedLinkedQueue;
//import EDU.oswego.cs.dl.util.concurrent.Semaphore;

import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Semaphore;

/**
 * 写线程
 *
 * @author filon51
 * @since 2014-02-14
 */
public class WriterThread extends Thread {
	private BusinessService m_businessService = null;
	private LinkedBlockingDeque m_writeDataQueue = null;
	private Semaphore m_semaphoreFinish = null;
	private int m_nProcessThreadCount = 0;
	private Logger m_log = Logger.getLogger(WriterThread.class);

	public WriterThread(BusinessService businessService,
						LinkedBlockingDeque writeDataQueue, Semaphore semaphoreFinish,
                        int processThreadCount) {
		m_businessService = businessService;
		m_writeDataQueue = writeDataQueue;
		m_semaphoreFinish = semaphoreFinish;
		m_nProcessThreadCount = processThreadCount;
	}

	public void run() {
		HashMap hashResult = null;
		int nCount = 0;
		m_log.debug("WriterThread start");

		while (true) {
			try {
				hashResult = (HashMap) m_writeDataQueue.take();
			} catch (Exception ex) {
				m_log.error("m_writeDataQueue.take()", ex);
				ex.printStackTrace();
			}

			if (hashResult == MultiThreadExecutor.getOBJ_WRITE_GUARD()) {
				nCount++;

				// 所有的处理线程都已经终止了，不会再提供数据了。
				if (nCount == m_nProcessThreadCount) {
					try {
						m_businessService.flushAllElements();
					} catch (Throwable e){
						m_log.error("==============WriterThread 调用 flushAllElements  发生异常 ！！!"+e);
						e.printStackTrace();
						m_businessService.setExceptionFlag(true);
						break;
					}
					break;
				} else {
					m_log.debug("nCount is " + String.valueOf(nCount)
							+ " and m_nProcessThreadCount is "
							+ String.valueOf(m_nProcessThreadCount));
				}
			}else if(hashResult == MultiThreadExecutor.getOBJ_EXCEPTION_GUARD()){//有子线程出现异常
				//异常回滚数据处理
				m_log.error("one dp Occured Exception!");
				m_businessService.dealException();
				nCount++;

				// 所有的处理线程都已经终止了，不会再提供数据了。
				if (nCount == m_nProcessThreadCount) {
					try {
						m_businessService.flushAllElements();
					} catch (Throwable e) {
						m_log.error("==============WriterThread 调用 flushAllElements  发生异常 ！！!"+e);
						e.printStackTrace();
						m_businessService.setExceptionFlag(true);
						break;
					}
					break;
				} else {
					m_log.info("nCount is " + String.valueOf(nCount)
							+ " and m_nProcessThreadCount is "
							+ String.valueOf(m_nProcessThreadCount));
				}
			}else {
				m_businessService.saveOneElement(hashResult);
			}
		}

		try {
			m_semaphoreFinish.release();
		} catch (Exception ex) {
			m_log.error("m_semaphoreFinish.release()", ex);
			ex.printStackTrace();
		}

		m_log.debug("WriterThread end");
	}
}
